package kafkaDemo;

import java.util.Properties;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
public class KafkaProducetest {
	 public static void main(String[] args) {
	        System.out.println("begin produce");
	        connectionKafka();
	        System.out.println("finish produce");
	    }

	    public static void connectionKafka() {
	        Properties props = new Properties();
	        /**
	         *  （用于建立与kafka集群连接的host/port组。数据将会在所有servers上均衡加载，不管哪些server是指定用于bootstrapping。这个列表仅仅影响初始化的hosts（用于发现全部的servers）。这个列表格式：
			*	host1:port1,host2:port2,...
			*	因为这些server仅仅是用于初始化的连接，以发现集群所有成员关系（可能会动态的变化），这个列表不需要包含所有的servers（你可能想要不止一 个server，尽管这样，可能某个server宕机了）。如果没有server在这个列表出现，则发送数据会一直失败，直到列表可用。）
	         */
	        props.put("bootstrap.servers", "172.17.6.10:9092,172.17.6.11:9092,172.17.6.12:9092");// kafka
	        
	        props.put("zookeeper.connect", "172.17.6.10:2181,172.17.6.11:2181,172.17.6.12:2181");//声明zk 
	      /**
	       * （producer需要server接收到数据之后发出的确认接收的信号，此项配置就是指procuder需要多少个这样的确认信号。此配置实际上代表了数据备份的可用性。以下设置为常用选项：
			*（1）acks=0： 设置为0表示producer不需要等待任何确认收到的信息。副本将立即加到socket  buffer并认为已经发送。没有任何保障可以保证此种情况下server已经成功接收数据，同时重试配置不会发生作用（因为客户端不知道是否失败）回 馈的offset会总是设置为-1；
			*（2）acks=1： 这意味着至少要等待leader已经成功将数据写入本地log，但是并没有等待所有follower是否成功写入。这种情况下，如果follower没有成功备份数据，而此时leader又挂掉，则消息会丢失。
			*（3）acks=all： 这意味着leader需要等待所有备份都成功写入日志，这种策略会保证只要有一个备份存活就不会丢失数据。这是最强的保证。
			*（4）其他的设置，例如acks=2也是可以的，这将需要给定的acks数量，但是这种策略一般很少用。）
	       */
	        props.put("acks", "all");
	       props.put("group.id", "tsxConsumer2");//必须要使用别的组名称， 如果生产者和消费者都在同一组，则不能访问同一组内的topic数据  
	        props.put("retries", 0);
	        props.put("batch.size", 16384);
	        props.put("linger.ms", 1);
	        props.put("buffer.memory", 33554432);
	     //   props.put("compression.type", "none");// 用于压缩数据的压缩类型。默认是无压缩。正确的选项值是none、gzip、snappy。 压缩最好用于批量处理，批量处理消息越多，压缩性能越好。
	      // props.put("client.id", "172.17.6.13"); 标明数据入库请求来源
	        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
	        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

	      
	        	 /**
	        	  * key 值可以做的工作： 依据key不同做响应的业务逻辑判断等
	        	  * 因此，如果消息前后有顺序关系，那么可以将先后顺序的消息Publish到同一个Partition，每个Consumer线程也是从一个Partition读取。
	        	  **并且，如果只有一个Partition的情况下，即使使用多个线程读取，实际上也只有一个thread会consume message
	        	  *因此Consumer的数量是根据Partition来制定，Consumer不应该多于Partition，否则多出的consumer线程会处于空闲状态
	        	  * 
	        	  */
	        Producer<String, String> producer = new KafkaProducer<>(props);
	                 for (int i = 0; i < 1000; i++) {
	                	 producer.send(new ProducerRecord<String, String>("gagaUserSignup",2, "echo gagahi", "唐思想你好"+i));
						
					}
	                 producer.close();
	    }
}
